package com.ch1.kafka001.commit;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

//kafka-topics.bat --zookeeper localhost:2181 --create --topic topic1 --replication-factor 1 --partitions 4
//消费者：同步和异步的组合提交偏移量，使用生产者： ProducerCommit
public class SyncAndAsync {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers","127.0.0.1:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"group01");

        props.put("enable.auto.commit",false);  //取消自动提交

        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);

        try{

            consumer.subscribe(Collections.singletonList("topic1"));

            while(true){
                ConsumerRecords<String,String> records = consumer.poll(500);

                for(ConsumerRecord <String,String> record: records){
                    System.out.println("主题："+record.topic()+"  分区："+record.partition()+
                            "  偏移量："+record.offset()+" key:"+record.key()+"  value:"+record.value());
                }

                consumer.commitAsync(); // 异步提交
            }

        }finally {
            try{
                consumer.commitSync(); // 为了万无一失，需要同步提交下
            }finally {
                consumer.close();
            }
        }
    }
}
